package cn.tedu.charging.order.mqtt;


import cn.tedu.charging.common.pojo.jsonResult.JsonResult;
import cn.tedu.charging.common.pojo.jsonResult.param.ChargingProcessParam;
import cn.tedu.charging.common.pojo.jsonResult.util.JsonUtils;
import cn.tedu.charging.common.pojo.jsonResult.vo.ChargingProcessVO;
import cn.tedu.charging.order.common.ChargingConstants;
import cn.tedu.charging.order.common.MqttConstants;
import cn.tedu.charging.order.common.OrderStatusConstants;
import cn.tedu.charging.order.pojo.dto.ChargingProcessDto;
import cn.tedu.charging.order.pojo.dto.ChargingResultDto;
import cn.tedu.charging.order.pojo.po.ChargingOrderFailPO;
import cn.tedu.charging.order.pojo.po.ChargingOrderSuccessPO;
import cn.tedu.charging.order.pojo.po.ChargingProcessPO;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;

/**
 * 用来消费 设备发送给订单服务的消息
 * 1 订阅 topic
 * 2 消费 消息
 */

@Slf4j
public class MqttConsumer implements MqttCallbackExtended {
   // private static final Logger log = LoggerFactory.getLogger(MqttConsumer.class);
    //使用构造器 注入mqttClient
    private MqttClient mqttClient;
    private MqttContext mqttContext;
    public MqttConsumer(MqttClient mqttClient,MqttContext mqttContext) {
        this.mqttClient = mqttClient;
        this.mqttContext = mqttContext;
    }

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.debug("连接成功,reconnect:{},serverURI:{}",reconnect,serverURI);
        //和EMQX连接成功之后，订阅topic-->设备返回的结果-->开始充电
        // 1 设备返回的 开始充电 结果  TOPIC_CHARGING_RESULT
        // 2 设备同步的充电进度  TOPIC_CHARGING_PROCESS
        String topicChargingResult = MqttConstants.TOPIC_CHARGING_RESULT;
        String topicChargingProcess = MqttConstants.TOPIC_CHARGING_PROCESS;
            try {
                log.debug("订单服务 订阅 设备返回的 <<开始充电结果>> Topic:{}",topicChargingResult);
                mqttClient.subscribe(topicChargingResult);
                log.debug("订单服务 订阅 设备返回的 <<开始充电结果>> Topic:{},成功",topicChargingResult);
                log.debug("订单服务 订阅 设备同步的 <<充电进度>> Topic:{}",topicChargingProcess);
                mqttClient.subscribe(topicChargingProcess);
                log.debug("订单服务 订阅 设备同步的 <<充电进度>> Topic:{},成功",topicChargingProcess);


            } catch (MqttException e) {
                throw new RuntimeException(e);
            }
    }


    @Override
    public void connectionLost(Throwable cause) {
        log.debug("连接断开,cause:{}", cause);

    }

    /**
     * 消息到达
     * @param  topic
     * @param  message
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        log.debug("消息到达,topic:{},message:{}" ,topic,message);
        if (MqttConstants.TOPIC_CHARGING_RESULT.equals(topic)) {
            log.debug("处理设备返回开始充电结果");
            handleChargingResult(message);
        }else if (MqttConstants.TOPIC_CHARGING_PROCESS.equals(topic)) {
            log.debug("处理设备同步的充电进度");
            handleChargingProcess(message);
        }

    }

    /**
     * 处理设备同步的充电进度
     * @param message
     */
    private void handleChargingProcess(MqttMessage message) {
        ChargingProcessDto chargingProcessDto = null;
        try{
            chargingProcessDto =
                    JsonUtils.fromJson(message.toString(), ChargingProcessDto.class);
        }catch (Exception e) {
            log.error("解析设备同步的消息失败,message:{}", message,e);
        }
        log.debug("收到设备同步的充电进度:{}",chargingProcessDto);
        if (chargingProcessDto != null) {
            ChargingProcessVO chargingProcessVO = getChargingCost(chargingProcessDto);
            saveChargingProcess(chargingProcessDto,chargingProcessVO);

        }
    }
    /**
     * 保存充电进度数据
     * 保存完整的数据到ES
     *
     * @param chargingProcessDto 设备同步的数据
     * @param chargingProcessVO 计算完价格后的数据
     */
    private void saveChargingProcess(ChargingProcessDto chargingProcessDto,
                                     ChargingProcessVO chargingProcessVO) {
        ChargingProcessPO chargingProcessPO = new ChargingProcessPO();
        BeanUtils.copyProperties(chargingProcessDto,chargingProcessPO);
        BeanUtils.copyProperties(chargingProcessVO,chargingProcessPO);
        //设置id 设备同步的时间
        chargingProcessPO.setId(System.currentTimeMillis());
        log.debug("保存充电进度数据到ES:{}",chargingProcessPO);
        mqttContext.getChargingProcessESRepository().save(chargingProcessPO);
        log.debug("保存充电进度数据到ES:{},成功",chargingProcessPO);
    }

    /**
     * 调用计价服务计算价格
     * @param chargingProcessDto
     * @return
     */
    private  @Nullable ChargingProcessVO  getChargingCost(ChargingProcessDto chargingProcessDto) {
        ChargingProcessParam chargingProcessParam = new ChargingProcessParam();
        //通过BeanUtils.copyProperties()复制属性
        BeanUtils.copyProperties(chargingProcessDto,chargingProcessParam);
//            chargingProcessParam.setOrderNo(chargingProcessDto.getOrderNo());
//            chargingProcessParam.setGunId(chargingProcessDto.getGunId());
        log.debug("调用计价服务计算价格入参:{}",chargingProcessParam);
        ChargingProcessVO chargingProcessVO = getCost(chargingProcessParam);
        log.debug("调用计价服务计算价格入参:{},出参:{}",chargingProcessParam,chargingProcessVO);
        return chargingProcessVO;
    }

    /**
     * 通过feign客户端 调用计价服务 获取设备的充电价格
     * @param chargingProcessParam
     * @return
     */
    private ChargingProcessVO getCost(ChargingProcessParam chargingProcessParam) {
        JsonResult<ChargingProcessVO> result= mqttContext.getCostClient().process(chargingProcessParam);
        if (result != null) {
            return result.getData();

        }
        return null;
    }

    /**
     * 处理设备返回的开始充电结果
     *
     * @param message
     */

    private void handleChargingResult(MqttMessage message) {
        ChargingResultDto chargingResultDto =null;
        try {
            chargingResultDto=
                    JsonUtils.fromJson(message.toString(),ChargingResultDto.class);
        }catch (Exception e){
            log.error("解析设备返回的消息失败,message:{}", message,e);
        }
        if (chargingResultDto!=null){
            String result = chargingResultDto.getResult();
            if (ChargingConstants.CHARGING_RESULT_SUCCESS.equals(result)){
                log.debug("订单服务 接收到设备返回的开始充电结果,开始充电成功",result);
                ChargingOrderSuccessPO chargingOrderSuccessPO = new ChargingOrderSuccessPO();
                chargingOrderSuccessPO.setBillId(chargingResultDto.getOrderNo());
                chargingOrderSuccessPO.setGunId(chargingResultDto.getGunId());
                chargingOrderSuccessPO.setBillStatus(OrderStatusConstants.ORDER_STATUS_CHARGING);
                log.debug("保存成功订单记录,chargingOrderSuccessPO:{}",chargingOrderSuccessPO);
                int row = mqttContext.getOrderSuccessMapper().insert(chargingOrderSuccessPO);
                log.debug("保存成功订单记录,row:{}",chargingOrderSuccessPO,row);

            }else{
                log.debug("订单服务 接收到设备返回的开始充电结果,开始充电失败",result);
                ChargingOrderFailPO chargingOrderFailPO = new ChargingOrderFailPO();
                chargingOrderFailPO.setBillId(chargingResultDto.getOrderNo());
                chargingOrderFailPO.setFailDesc("设备有响应，开始充电失败");
                chargingOrderFailPO.setGunId(chargingResultDto.getGunId());
                log.debug("保存失败订单记录,chargingOrderFailPO:{}",chargingOrderFailPO);
                int row= mqttContext.getOrderFailMapper().insert(chargingOrderFailPO);
                log.debug("保存失败订单记录,row:{}",chargingOrderFailPO,row);
                log.debug("<<通知充电用户>> 更换设备");
                log.debug("<<通知充电用户>> 给用户发短信，表示歉意");
                log.debug("<<通知充电用户>> 给用户发短信，发送优惠券 营销系统，给用户心灵上的安慰");
                log.debug("<<通知设备管理员>> 对设备进行检修");
                log.debug("<<修改枪的状态为 维护中>>");
            }
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("消息发送完成,token:{}",token);

    }
}
